package com.zero.stream.map;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.zero.analy.PindaoRD;
import com.zero.input.KafkaMessage;
import com.zero.log.UserscanLog;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * 频道热点map
 */
public class PindaoRDMap extends RichMapFunction<PindaoRD, PindaoRD> {

    private transient ValueState<Long> currentTotalCount;

    @Override
    public PindaoRD map(PindaoRD value) throws Exception {
        Long totalCount = currentTotalCount.value();
        Long count = value.getCount();
        if (ObjectUtil.isNull(count)) {
            count = 0L;
        }
        totalCount += count;

        currentTotalCount.update(totalCount);
        PindaoRD pindaoRD = new PindaoRD();
        pindaoRD.setPingdaoid(value.getPingdaoid());
        pindaoRD.setCount(currentTotalCount.value());
        return pindaoRD;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        currentTotalCount = getRuntimeContext().getState(new ValueStateDescriptor<>("currentTotalCount", Long.class));
    }
}
